package my.thesis.bolts.persistors;

import java.sql.Connection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import my.thesis.analytics.metrics.LastTimePersistMetric;

import org.joda.time.DateTime;
import redis.clients.jedis.Jedis;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class TextStreamPersistor extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private Jedis jedis;
	private OutputCollector collector;
	Connection connect=null;
	
	transient LastTimePersistMetric _lastTimeMetric;
	private static Boolean metricsOn=false;
	
	private String database;
	
	public TextStreamPersistor (String connectionString)
	{
		this.database=connectionString;
	}

	 public void declareOutputFields(OutputFieldsDeclarer declarer) 
	 {
		    // No output fields to be declared
	 }
	  
	 public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)  
	  {
		this.collector=collector;
	    //jedis = new Jedis(this.database);
	    jedis=new Jedis(database,6379,60);
	    jedis.select(4);
	    
	    if(metricsOn)
	    {
	    	_lastTimeMetric=new LastTimePersistMetric();
	    	context.registerMetric("sentimentPersist", _lastTimeMetric, 40);
	    	_lastTimeMetric.setStartTime();
	    }
	  }

	  public void execute(Tuple tuple) 
	  {
		  if(metricsOn)
			 {
				 _lastTimeMetric.setTimeReceived();
			 }
		 
		 	 @SuppressWarnings("unchecked")
		 	 ConcurrentHashMap<String,Integer> scores =  (ConcurrentHashMap<String,Integer>) tuple.getValueByField("countsTotal");
		     
		 	 //TEST
		     System.out.println("Persisting score...");
		     
		     Set<String> keys = new HashSet<String>();
		     
		     for(String h : scores.keySet())
		     {	
		    	 String key=new String(h);
		    	 keys.add(key);
		     }
		     
		     for(String k : keys)
		     {
		    	 try
		    	 {
		    		 jedis.hincrBy("SentimentByCountry",k,scores.get(k));
		    	 }
		    	 catch(Exception e)
		    	 {
		    		 System.out.println("Error saving to redis...");
		    	 }
		    	 
		     }
	  }

	  public void cleanup() {
	    if (jedis.isConnected()) 
	    {
	        jedis.quit();
	    }
	  }
}
